package org.pz.mq.demo;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.pz.entity.UserEntity;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;

@Component
@Slf4j
public class MqDemoCustomer {

    @RabbitListener(queues = MqDemoConfig.direct_queue1)
    public void d1(String message1, Message message, Channel channel) throws IOException {
        log.info( "消费者 {} 消费信息：{}", MqDemoConfig.direct_queue1, message1);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(queues = MqDemoConfig.direct_queue2)
    public void d2(String message1, Message message, Channel channel) throws IOException {
        log.info( "消费者 {} 消费信息：{}", MqDemoConfig.direct_queue2, message1);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(queues = MqDemoConfig.direct_queue3)
    public void d3(UserEntity user, Message message, Channel channel) throws IOException {
        log.info( "消费者 {} 消费信息：{}", MqDemoConfig.direct_queue3, user);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(queues = MqDemoConfig.fanout_queue1)
    public void f1(String message1, Message message, Channel channel) throws IOException {
        log.info( "消费者 {} 消费信息：{}", MqDemoConfig.fanout_queue1, message1);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(queues = MqDemoConfig.fanout_queue2)
    public void f2(String message1, Message message, Channel channel) throws IOException {
        log.info( "消费者 {} 消费信息：{}", MqDemoConfig.fanout_queue2, message1);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(queues = MqDemoConfig.fanout_queue3)
    public void f3(String message1, Message message, Channel channel) throws IOException {
        log.info( "消费者 {} 消费信息：{}", MqDemoConfig.fanout_queue3, message1);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    /**
     * 这是个错误的使用示例，模拟消费期间出现异常
     * 1. 如果是自动确认模式，
     *      不需要 channel.basicAck()，
     *      会在耗尽重试次数之后，抛出这个异常，
     *      消息被自动确认，
     *      根据配置的 MessageRecoverer 来处理消息
     * 2. 如果是手动确认模式，
     *      这里就是有问题的，因为无法执行到 channel.basicAck() 而无法确认消息，后续可能导致消费者阻塞。
     *      其他都跟自动确认模式下一样
     * */
    @RabbitListener(queues = MqDemoConfig.retry_queue)
    public void r1(String message1, Message message, Channel channel) throws IOException {
        log.info("消费者 {} 消费消息 {} 了！！！", MqDemoConfig.retry_queue, message1);
        //模拟一个异常，
        System.out.println(1/0);
        //确认消费
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(queues = MqDemoConfig.retry_failure_queue)
    public void r2(String message1, Message message, Channel channel) throws IOException {
        log.info("消费者 {} 消费消息 {} 了！！！", MqDemoConfig.retry_failure_queue, message1);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(queues = MqDemoConfig.demo_queue)
    public void de1(String message1, Message message, Channel channel) throws IOException {
        log.info("消费者 {} 消费消息 {} 了！！！", MqDemoConfig.demo_queue, message1);
        //拒绝并不重新入列，会进入死信队列
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
    }
}
